package cn.com.bluemoon.bd.flink.dim;

import cn.com.bluemoon.bd.common.spring.SpringContextHolder;
import cn.com.bluemoon.bd.common.spring.redis.SimpleRedisTemplate;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.jdbc.core.JdbcTemplate;

import java.io.Serializable;
import java.util.List;

public class DimDataLoadFunc extends ProcessFunction<String, String> implements Serializable {
    private transient JdbcTemplate jdbcTemplate;
    private transient SimpleRedisTemplate redisTemplate;
    private List<IDimLoader> loaders = Lists.newArrayList();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.jdbcTemplate = SpringContextHolder.getBean("jdbcTemplate", JdbcTemplate.class);
        this.redisTemplate = SpringContextHolder.getBean("simpleRedisTemplate", SimpleRedisTemplate.class);
    }

    @Override
    public void processElement(String element, Context context, Collector<String> collector) throws Exception {
        this.loaders.forEach(loader -> loader.load(this.jdbcTemplate, this.redisTemplate));
    }

    public void add(IDimLoader loader) {
        this.loaders.add(loader);
    }

}
